/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * htap_flush.h
 * API for the HTAP Flush worker
 *
 * IDENTIFICATION
 * src/include/storage/htap/htap_flush_worker.h
 *
 * -------------------------------------------------------------------------
 */

#ifndef HTAP_FLUSH_WORKER_H
#define HTAP_FLUSH_WORKER_H

#include "knl/knl_thread.h"
#include "htap_common.h"
#include "gs_thread.h"
#include <mutex>
#include "storage/htap/htap_delta_table.h"
#include "access/imcs/imcu.h"
#include "access/imcs/imcs_flush_insert.h"

namespace htap {

class HTAPFlushCtx : public BaseObject {
public:
    HTAPFlushCtx(int rel_oid, CommitSeqNo csn)
        : m_htap_delta(NULL),
          m_rel_oid(rel_oid),
          m_table(NULL),
          m_rel(NULL),
          parent_rel(NULL),
          part(NULL),
          m_imcs_desc(NULL),
          csn(csn),
          m_tuple_desc(NULL),
          m_imcs_to_values(NULL),
          m_result_rel_info(NULL),
          m_new_rgs(0),
          m_num_rows(0),
          m_insert_time(0),
          m_num_rgs(0),
          m_num_filtered_rgs_1(0),
          m_num_filtered_rgs_2(0),
          m_delete_time(0),
          m_start_time(0),
          m_end_time(0),
          m_deleted(0)

    {
        initRefCount();
    }

    ~HTAPFlushCtx() {}
    void Destroy();

    void GetDeltaTable(LOCKMODE lockmode);
    bool setup();
    void ReleaseDeltaTable(LOCKMODE lockmode);

    /** @var Delta Extractor */
    HTAPDelta *m_htap_delta;

    /** @var rel oid of the current table */
    int m_rel_oid;

    /** @var mot::table of the current table */
    MOT::Table *m_table;

    /** @var Relation object of the current table */
    Relation m_rel;

    /** @var Parent Relation object of the current table, if partitioned */
    Relation parent_rel;

    /** @var Partition object of the current table, if partitioned */
    Partition part;

    /** @var IMCSDesc of the current table */
    IMCSDesc *m_imcs_desc;

    /** @var Flush csn */
    CommitSeqNo csn;

    /** @var Table tuple desc */
    TupleDesc m_tuple_desc;

    /** @var imcs_to_values for this table */
    int2 *m_imcs_to_values;

    /** @var rel_info for this table */
    ResultRelInfo *m_result_rel_info;

    /** @var new_rgs (to update once at the end of Flush) */
    uint32 m_new_rgs;

    // Statistics for each task
    int32 m_num_rows;
    int64 m_insert_time;
    int32 m_num_rgs;
    int32 m_num_filtered_rgs_1;
    int32 m_num_filtered_rgs_2;
    int64 m_delete_time;
    int64 m_start_time;
    int64 m_end_time;
    int32 m_deleted;

    void initRefCount()
    {
        std::lock_guard<std::mutex> guard(m_lock);
        m_ref_count = 0;
    };
    uint16_t getRefCount()
    {
        std::lock_guard<std::mutex> guard(m_lock);
        return m_ref_count;
    };
    void updateRefCount(int8_t count)
    {
        std::lock_guard<std::mutex> guard(m_lock);
        m_ref_count += count;
    };

private:
    /** @var ref_count */
    uint16_t m_ref_count;

    /** @var Lock to serialize thread state management. */
    std::mutex m_lock;
};
class HTAPFlushWorker {
public:
    HTAPFlushWorker() : m_need_to_stop(false) {}

    /** @brief Destructor. */
    ~HTAPFlushWorker() {}

    /* *
     * @brief Launches the flush thread.
     * @param threadId The resulting flush thread identifier.
     * @return @ref HTAP_RC_OK if succeeded, otherwise error code.
     */
    ThreadId Start(int thread_num);

    /* *
     * @brief Orders the flush thread to stop. This is a blocking call that waits until the flush thread stops.
     * @return @ref HTAP_RC_OK if succeeded, otherwise error code.
     */
    RC Stop();

    /** @brief Queries whether the flush thread is currently running. */
    bool IsRunning();


    /** @brief Retrieves the flush thread id. */
    ThreadId GetThreadId();

    /** @brief Executes the flush thread's main loop. */
    void RunFlushWorker();

    bool SignaledToStop(uint64_t timeoutMillis);

private:
    void InitThread();

    char m_user_name[NAMEDATALEN];

    int m_thread_num;

    /** @var Lock to serialize thread state management. */
    std::mutex m_lock;

    /** @var Flush thread identifier. */
    ThreadId m_flushWorkerThread;

    /** @var Cycle Memory controller - created and released each cycle */
    MemoryContext m_flush_ctx;

    /** @var Flush thread state. */
    bool m_isRunning;

    bool m_need_to_stop;

    /** @var Struct holding table context variables */
    HTAPFlushCtx *m_table_ctx;

    IMCSFlushInsert *m_imcs_insert;

    /** @brief Main thread function */
    void PerformTasks(MOT::TxnManager *txn);

    IMCU *getCU(IMCUDesc *cu_desc_ptr, int col_idx, CacheSlotId_t &slot_id);

    void handleDeletes(int rg_idx);

    IMCUDesc **initImcuDescsArray(int rg_idx);

    void addToRowgroups(HTAPDelta::ScanContext &scan_ctx, bool multithread, int free_rows = IMCS_MAX_ROWS_SIZE,
        IMCU **old_cus = NULL, IMCUDesc **imcu_descs_array = NULL);
    RC getOldCUs(IMCU **old_cus, TupleDesc tuple_desc_for_old_cus, IMCUDesc **imcu_descs_array,
        CacheSlotId_t *slot_array);
    void completeInserts(HTAPDelta::ScanContext &scan_ctx, int complete_rg_idx, int free_rows);
    void compact(int rg);
    void handleCompactForNullCu(HTAPDeleteMap *delMap, IMCUDesc *cu_desc);
    void updateCompactIMCS(IMCSDesc *imcs_desc, int rg);
    void updateCompactCu(IMCU *cu, IMCUDesc *cu_desc, HTAPDeleteMap *delMap, Oid type_oid, int attlen, int atttypid);
    void setMinMax(IMCUDesc *cu_desc, bool *first, Oid atttypid, int attlen, char *val_addr);
    int handleCompactNulls(IMCU *cu, IMCUDesc *cu_desc, HTAPDeleteMap *delMap, char *old_nulls);

    CUUncompressedRetCode getCuDataFromRemote(IMCUDesc *cu_desc_ptr, IMCU *cu_ptr, int col_idx, int val_size,
        const int &slot_id, IMCUStorage *cu_storage);
    void checkConsistenceOfCuData(IMCUDesc *cu_desc_ptr, IMCU *cu, AttrNumber col) const;
};
} // namespace htap
#endif // HTAP_FLUSH_WORKER_H
